package com.ruyuan.dfs.namenode.editslog;

import com.ruyuan.dfs.common.utils.FileUtil;
import com.ruyuan.dfs.namenode.config.NameNodeConfig;
import com.ruyuan.dfs.namenode.fs.PlaybackEditLogCallback;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.StopWatch;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * 表示 EditLogs
 *
 * @author Sun Dasheng
 */
@Slf4j
public class FsEditLog {

    //提前编译好 正则表达式 提高性能
    private static Pattern indexPattern = Pattern.compile("(\\d+)_(\\d+)");

    //配置类
    private NameNodeConfig nameNodeConfig;

    /**
     * 每条editLog的id，自增
     */
    private volatile long txIdSeq = 0;

    /**
     * 双缓冲
     */
    private DoubleBuffer editLogBuffer;

    /**
     * 每个线程保存的txid
     */
    private ThreadLocal<Long> localTxId = new ThreadLocal<>();

    /**
     * 当前刷新磁盘最大的txId
     */
    private volatile long syncTxid = 0;

    /**
     * 当前是否在刷磁盘
     */
    private volatile boolean isSyncRunning = false;

    /**
     * 是否正在调度一次刷盘的操作
     * 此处变量来 控制是否可以向  editLogBuffer(内存) 写入数据
     * 因为 editLogBuffer 可能满了 会执行 交换缓冲区操作 此时 是不可以向 editLogBuffer(内存) 写入数据的
     * 需要等待 交换完毕 才可以继续向 editLogBuffer(内存) 写入数据
     */
    private volatile Boolean isSchedulingSync = false;

    /**
     * 需要经常删除不必要的数据 以防内存泄露
     * 已经刷入磁盘中的 editLog 文件, 升序
     * 相当于索引文件
     * CopyOnWriteArrayList 来减少并发问题
     */
    private List<EditslogInfo> editLogInfos = null;

    public FsEditLog(NameNodeConfig nameNodeConfig) {
        //NameNode配置
        this.nameNodeConfig = nameNodeConfig;
        //双缓冲机制
        this.editLogBuffer = new DoubleBuffer(nameNodeConfig);
        //从磁盘中加载所有的 editsLog 文件信息
        this.loadEditLogInfos();
    }


    /**
     * 写入一条 editLog 同步模式
     *
     * @param editLog 内容
     */
    public void logEdit(EditLogWrapper editLog) {
        //同步阻塞 获得锁  并发下逐一获得锁   是否可以优化 ???
        synchronized (this) {
            // 刚进来就直接检查一下是否 可以向 editLogBuffer(内存) 中添加数据
            waitSchedulingSync();//1. 改动 : 这里的判断就可以进入 条件队列

            //操作序列号递增
            txIdSeq++;//2. 改动 : 这里可以从客户端生成

            // 保存 txId 到当前线程中
            long txid = txIdSeq;
            //当前线程中的 消息ID 变量
            localTxId.set(txid);

            // 构造一条 editLog 写入缓冲区
            editLog.setTxId(txid);
            try {
                //先把数据存放在内存 buffer 中
                //这里万一出现 溢出怎么办？不会溢出，会自动扩容的
                editLogBuffer.write(editLog);
            } catch (IOException e) {
                e.printStackTrace();
            }

            //每次不用刷盘 直到 到了一定容量才去 刷盘
            //上一步填充完了以后 再去判断是否满了需要再去刷盘 (半同步+半异步 刷盘策略)
            if (!editLogBuffer.shouldForceSync()) {
                //释放锁 下一个线程可以进来
                return;
            }

            // 如果有一个线程进行到这里，就说明内存中存储的数据量 需要刷进磁盘了
            //设置成 true 后  那么执行到第 95 行代码的线程  就开始在内存中 循环阻塞了
            isSchedulingSync = true; //3. 改动 ： 阻塞条件队列线程 使其 不能向 buffer 中填充数据
        }//释放锁

        // 同步刷磁盘 走到这里说明就是当前线程已经把数据写到 buffer 中了 需要刷盘
        // 可能会并发执行 1 2 3 4 个线程 同时到达
        logSync();
    }

    /**
     * 等待正在调度的刷磁盘的操作
     */
    private void waitSchedulingSync() {
        try {
            //刚开始写文件数据时是不需要进行刷盘的 因此这里也就直接退出了
            //但是经过一段时间的 currentBuffer 的写入后 那么就需要进行 缓冲区刷盘了
            //因此在 114 行代码中 设置了为 true ，那么其余的线程就将进入多次不断的循环阻塞中
            //那么什么时机会被唤醒呢？
            //189行代码会被唤醒
            while (isSchedulingSync) {
                // 释放锁了 那么此时就能有其他线程进入代码块了
                // 此时就释放锁，等待一秒 , 自动被唤醒后 , 再次尝试获取锁，获取到锁时再去判断是否可写
                // isSchedulingSync是否为false，就可以脱离出while循环  否则后来进来的线程依然被阻塞主
                wait(1000);
            }
        } catch (Exception e) {
            log.info("waitSchedulingSync has interrupted !!");
        }
    }

    /**
     * 需要进行 同步刷磁盘
     */
    private void logSync() {
        //同步阻塞 获取锁 1 2 3 4 线程并发执行
        synchronized (this) {
            //获得当前要执行刷盘的线程的 消息ID
            long currentThreadTxId = localTxId.get();
            localTxId.remove();
            /**
             * 比如：当前 syncTxid =1 ，来了 txID=2, 3 ,4： 4先进行刷盘，那么 2 ，3 被唤醒后，也就不再进行刷盘了
             * 在这种情况下当前现线程需要等待：
             * 1. 有其他线程正在刷磁盘，但是其他线程刷的磁盘的最大 txId 比当前需要刷磁盘的线程id少
             * 这通常表示：正中在刷磁盘的线程不会把当前线程需要刷的数据刷到磁盘
             */
            //条件一：currentThreadTxId > syncTxId  成立：相当于第二波刷盘来了 但是 第一波还没结束 所以就不停的 阻塞
            //                                  不成立：说明当前的线程的任务已经过期 直接退出
            //条件二：isSyncRunning  成立：说明已经有线程马上执行刷盘了，说明什么？ 等下再刷盘？
            //                                  不成立：没有线程在刷盘 默认是 false
            while (currentThreadTxId > syncTxid && isSyncRunning) {
                try {
                    //释放锁  等待1秒后 唤醒 唤醒后 发现还在 同步 继续循环阻塞
                    wait(1000);//4. 改动 : 刷盘线程执行 阻塞
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            /**
             * 多个线程在上面等待，当前一个线程刷磁盘操作完成后，唤醒了一堆线程(不确定性)，此时只有一个线程获取到锁
             * 这个线程会进行刷磁盘操作，当这个线程释放锁之后，其他被唤醒的线程会依次获取到锁
             * 此时每个被唤醒的线程需要重新判断一次，自己要做的事情是不是被其他线程干完了
             */
            if (currentThreadTxId <= syncTxid) {
                //直接退出 释放锁 因为该干的活儿 被已经干完了 让下一个线程进入代码块
                return;
            }

            // 需要进行刷盘 -> 交换两块缓冲区  -> currentBuffer 缓冲区就又可以进行缓存数据了(对应第100行代码)
            editLogBuffer.setReadyToSync();

            // 记录最大的 currentThreadTxId
            syncTxid = currentThreadTxId;

            // 此时 已经交换完 缓冲区 因此 其他线程可以向 buffer 中 填充数据了
            // 当这里 又赋值成 false 后，那么处于 128 行代码就有机会跳出循环了
            //但是 此处 将 刷盘 跟 填充buffer 强耦合了 应该 解耦 互相不影响才行
            isSchedulingSync = false;//3. 改动 ： 唤醒相关条件队列 向 buffer 中填充数据

            // 唤醒哪些正在wait()的线程 比如：132 行 进行定时阻塞的线程 或者 161 行代码 唤醒没作用 醒了也是继续睡眠
            //又由于上面 181 行 代码 进行了缓冲区置换，因此从128 行 while 循环代码中唤醒过来的线程可以继续 向 currentBuffer 缓冲区中写入数据了
            //(只是唤醒沉睡的线程，而不会立即释放锁)
            notifyAll();// 意味着什么?  答 : 判断填充 buffer 的线程 | 判断 是否可以刷盘的线程 全唤醒了

            // 需要设置成正在刷磁盘
            // 那么上面的 后来的线程 就会阻塞主
            //currentThreadTxId > syncTxid 的线程就会阻塞主
            //currentThreadTxId <= syncTxid 的线程直接返回
            isSyncRunning = true;
        }//锁释放了  答 : 判断填充 buffer 的线程 | 判断 是否可以刷盘的线程 全又争抢锁了

        //这里释放锁 目的是啥 ? 答 : 是为了 刷盘 , 但是同一把锁可能会被 要写入buffer 的线程抢到 , 从而耽搁了刷盘线程,159行处的线程
        //因此 两个用途 应该两把锁来控制

        try {
            //把缓冲区的 editLog 数据刷新到磁盘 ，并没有 加锁 刷盘
            //并且返回 当前 512kb 的文件中的 最大|最小条目  保存到内存中
            //但是 需要重新计算文件名字  -> 执行打开文件句柄 -> 写入数据
            EditslogInfo editslogInfo = editLogBuffer.flush();
            //是进入磁盘的概括体
            //这里可以使用 CompleteFuture.complete() 异步编程
            if (editslogInfo != null) {
                //每次写入磁盘文件时 保存下来 为了之后的 BackUpNode 来 fetch editLog 日志
                editLogInfos.add(editslogInfo);
            }
        } catch (IOException e) {
            log.info("FSEditLog 刷磁盘失败：", e);
        }

        //刷完盘的线程 再次尝试 获得锁 目的是 唤醒 172 行的线程 可以继续执行 刷盘了
        synchronized (this) {
            // 同步完了磁盘之后，就会将标志位复位，再释放锁
            // 此时 172 行代码就会跳出循环 进行刷盘判断
            isSyncRunning = false;//5. 改动 ：唤醒 刷盘 条件队列
            //唤醒哪些正在wait()的线程 比如：132 行  但是由于185 行的赋值 所以没有线程在132行阻塞 = 无效  ||  161行代码  会被 唤醒 走接下来的逻辑 有效
            notifyAll();//(只是唤醒沉睡的线程，而不会立即释放锁)
        }
    }

    /**
     * 关闭时 NameNode
     * 强制把内存缓冲里的数据刷入磁盘中
     */
    public void forceFlush() {
        //还是需要锁
        synchronized (this) {
            try {
                //交换两块缓冲区
                editLogBuffer.setReadyToSync();
                EditslogInfo editslogInfo = editLogBuffer.flush();
                //此时还有必要进行保存吗???
                //我认为既然已经关机了 没有必要进行保存了
                if (editslogInfo != null) {
                    editLogInfos.add(editslogInfo);
                }
            } catch (IOException e) {
                log.error("强制刷新EditLog缓冲区到磁盘失败.", e);
            }
        }
    }

    /**
     * 获取当前写 editLog 的缓冲区
     *
     * @return 当前写 editLog 的缓冲区
     */
    public List<EditLogWrapper> getCurrentEditLog() {
        //加锁了
        synchronized (this) {
            return editLogBuffer.getCurrentEditLog();
        }
    }

    /**
     * NameNode 重启时回放 editLog 到内存中
     *
     * @param txiId    回放比 txId 大的日志
     * @param callback 回放日志回调
     * @throws IOException IO 异常
     */
    public void playbackEditLog(long txiId, PlaybackEditLogCallback callback) throws IOException {
        long currentTxSeq = txiId;
        this.txIdSeq = currentTxSeq;
        //获取比minTxId更大的editlog文件，经过排序后的文件
        List<EditslogInfo> sortedEditLogsFiles = getSortedEditLogFiles(txiId);
        StopWatch stopWatch = new StopWatch();
        for (EditslogInfo info : sortedEditLogsFiles) {
            if (info.getEnd() <= currentTxSeq) {
                continue;
            }
            // 从文件中读取EditLog
            List<EditLogWrapper> editLogWrappers = readEditLogFromFile(info.getName());
            stopWatch.start();
            for (EditLogWrapper editLogWrapper : editLogWrappers) {
                long tmpTxId = editLogWrapper.getTxId();
                if (tmpTxId <= currentTxSeq) {
                    continue;
                }
                currentTxSeq = tmpTxId;
                this.txIdSeq = currentTxSeq;
                if (callback != null) {
                    callback.playback(editLogWrapper);
                }
            }
            stopWatch.stop();
            log.info("回放editLog文件: [file={}, cost={} s]", info.getName(), stopWatch.getTime() / 1000.0D);
            stopWatch.reset();
        }
    }

    /**
     * 从文件中读取 EditLog
     *
     * @param absolutePath 绝对路径
     * @return EditLog
     * @throws IOException IO异常
     */
    public List<EditLogWrapper> readEditLogFromFile(String absolutePath) throws IOException {
        return EditLogWrapper.parseFrom(FileUtil.readBuffer(absolutePath));
    }


    /**
     * 获取比 minTxId 更大的 editLog 文件，经过排序后的文件
     * 比如磁盘中有文件：
     * <p>
     * 1_1000.log
     * 1001_2000.log
     * 2001_3000.log
     * <p>
     * 如果minTxId=1500，则会返回: [1001_2000.log, 2001_3000.log]
     *
     * @param minTxId 最小的 txId
     * @return 排序后的文件信息
     */
    public List<EditslogInfo> getSortedEditLogFiles(long minTxId) {
        List<EditslogInfo> result = new ArrayList<>();
        //遍历已经存盘的数据条目
        for (EditslogInfo editslogInfo : editLogInfos) {
            if (editslogInfo.getEnd() <= minTxId) {
                continue;
            }
            result.add(editslogInfo);
        }
        return result;
    }

    /**
     * 从磁盘中加载所有的 editsLog 文件信息
     */
    private void loadEditLogInfos() {
        //都是通过底层的数组的新副本来实现的
        this.editLogInfos = new CopyOnWriteArrayList<>();
        //获得系统元目录
        File dir = new File(nameNodeConfig.getBaseDir());
        //必须是文件目录形式
        if (!dir.isDirectory()) {
            return;
        }
        File[] files = dir.listFiles();
        if (files == null || files.length == 0) {
            return;
        }
        //遍历底层文件夹
        for (File file : files) {
            if (!file.getName().contains("edits")) {
                continue;
            }
            //获得每个 editLog 文件中的消息id值
            long[] index = getIndexFromFileName(file.getName());
            //首次启动 则需要 从磁盘中获取 已经刷盘了的文件 再次放入内存中
            this.editLogInfos.add(new EditslogInfo(index[0], index[1], nameNodeConfig.getBaseDir() + File.separator + file.getName()));
        }
        this.editLogInfos.sort(null);
    }


    /**
     * 从文件名中提取 index
     *
     * @param name 文件名  1_1000.log
     * @return 数组，例如：[1,1000]
     */
    private long[] getIndexFromFileName(String name) {//editslog-1_1.log
        Matcher matcher = indexPattern.matcher(name);
        long[] result = new long[2];
        if (matcher.find()) {
            result[0] = Long.parseLong(matcher.group(1));
            result[1] = Long.parseLong(matcher.group(2));
        }
        return result;
    }

    /**
     * 清除比 txId 小的日志文件
     * <p>
     * 比如磁盘中有文件：
     * <p>
     * 1_1000.log
     * 1001_2000.log
     * 2001_3000.log
     * <p>
     * 如果txid=1500，则会删除文件: [1_1000.log]
     *
     * @param txId txId
     */
    public void cleanEditLogByTxId(long txId) {
        log.info("清除比 txId:{} 小的日志文件", txId);
        List<EditslogInfo> toRemoveEditLog = new ArrayList<>();
        for (EditslogInfo editslogInfo : editLogInfos) {
            if (editslogInfo.getEnd() > txId) {
                continue;
            }
            File file = new File(editslogInfo.getName());
            if (file.delete()) {
                log.info("删除editLog文件: [file={}]", editslogInfo.getName());
            }
            toRemoveEditLog.add(editslogInfo);
        }
        log.info("清除比 txId:{} 小的日志文件,{},{}", txId, editLogInfos, toRemoveEditLog);
        editLogInfos.removeAll(toRemoveEditLog);
    }
}
